﻿using System;
using System.Net;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using Solution.Application.Modules;

namespace Solution.Application.Services.Imples
{
    public class KafkaService : IKafkaService
    {
        private readonly ProducerConfig config;

        public KafkaService(IOptions<KafkaOptions> options)
        {
            config = new ProducerConfig
            {
                //BootstrapServers = options.Value.Bootstrap,
                BootstrapServers = "PLAINTEXT://192.168.56.101:9092",
                Partitioner = Partitioner.Random,
                ClientId = Dns.GetHostEntry(Environment.MachineName).HostName,
                SecurityProtocol = SecurityProtocol.Plaintext
            };
        }

        public async Task ProducerAsync<TValue>(string topic, TValue value)
        {
            using (var producer = new ProducerBuilder<string, string>(config).Build())
            {
                try
                {
                    await producer.ProduceAsync(topic, new Message<string, string>()
                    {
                        Key = topic,
                        Value = JsonConvert.SerializeObject(value)
                        //Value = $"test-value {DateTime.Now.ToString("G")}"
                    });
                }
                catch (Exception ex)
                {
                }
            }
        }
    }
}